package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: DwsTradeSkuOrderWindow
 * Description:
 * sku粒度下单业务过程聚合统计
 * @Create 2024/4/27 9:09
 */
public class DwsTradeSkuOrderWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTradeSkuOrderWindow().start(
                10029,
                4,
                "dws_trade_sku_order_window",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaSourceDS) {
        //TODO 1.处理空消息（数据源是左外连接而成的，所以会产生重复数据。其实Flink自动会处理空消息），并对流中数据进行类型转换
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaSourceDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        if (!StringUtils.isEmpty(value)) {
                            JSONObject jsonObj = JSON.parseObject(value);
                            out.collect(jsonObj);
                        }
                    }
                }
        );
        //jsonObjDS.print();


        //TODO 2.去重
        //2.1 按照唯一键分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(j -> j.getString("id"));

        //2.2 去重方案1：状态 + 定时器（缺点，时效性差；优点：数据不膨胀）
        /*
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> stateProperties =
                                new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        lastJsonObjState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj,
                                               KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        //状态中获取上条数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj != null) {
                            //状态中没有数据，将数据存入状态，设置定时器等待重复数据更新状态
                            lastJsonObjState.update(jsonObj);

                            TimerService timerService = ctx.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);

                        } else {
                            //状态中有数据，当前为重复数据，判断谁的时间大，大的放到状态中（其实kafka中指定id为主键了，数据一定在同一分区，是可以保证有序的）
                            //伪代码，需要数据源携带当时的聚合处理时间
                            String curTs = jsonObj.getString("");
                            String lastTs = lastJsonObj.getString("");
                            if (curTs.compareTo(lastTs) >= 0) {
                                lastJsonObjState.update(jsonObj);
                            }
                        }

                    }

                    @Override
                    public void onTimer(long timestamp,
                                        KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx,
                                        Collector<JSONObject> out) throws Exception {
                        //定时器被触发后，从状态获取数据
                        JSONObject jsonObj = lastJsonObjState.value();
                        //传递数据
                        out.collect(jsonObj);
                        //清除状态
                        lastJsonObjState.clear();


                    }
                }
        );
        */

        //2.3 去重方案2：状态 + 抵消（度量取反；时效性好，但数据膨胀）
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> stateProperties =
                                new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);

                        stateProperties.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());

                        lastJsonObjState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj,
                                               KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        //从状态中获取数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj != null) {
                            //重复了，将状态中影响到度量值的字段进行取反，传递到下游
                            /*
                            "create_time":"2024-04-27 10:18:47",
                            "sku_num":"1",
                            "split_original_amount":"129.0000",
                            "split_coupon_amount":"0.0",
                            "sku_id":"27",
                            "date_id":"2024-04-27",
                            "user_id":"281",
                            "province_id":"31",
                            "sku_name":"索芙特i-Softto 口红不掉色唇膏保湿滋润 璀璨金钻哑光唇膏 Z02少女红 活力青春 璀璨金钻哑光唇膏 ",
                            "id":"39350",
                            "order_id":"23653",
                            "split_activity_amount":"0.0",
                            "split_total_amount":"129.0",
                            "ts":1714184327}
                            */
                            String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                            String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                            String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                            String splitTotalAmount = lastJsonObj.getString("split_total_amount");

                            lastJsonObj.put("split_original_amount", "-" + splitOriginalAmount);
                            lastJsonObj.put("split_coupon_amount", "-" + splitCouponAmount);
                            lastJsonObj.put("split_activity_amount", "-" + splitActivityAmount);
                            lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);

                            out.collect(lastJsonObj);

                        }

                        lastJsonObjState.update(jsonObj);

                        out.collect(jsonObj);

                    }
                }
        );


        //TODO 3.指定Watermark生成策略、提取时间时间字段
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = distinctDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts") * 1000;
                                    }
                                }
                        )
        );


        //TODO 4.jsonObj -> 实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> beanSD = withWatermarkDS.map(
                new MapFunction<JSONObject, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {

                        String skuId = jsonObj.getString("sku_id");
                        String splitOriginalAmount = jsonObj.getString("split_original_amount");
                        String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                        String splitActivityAmount = jsonObj.getString("split_activity_amount");
                        String splitTotalAmount = jsonObj.getString("split_total_amount");
                        Long ts = jsonObj.getLong("ts");

                        return TradeSkuOrderBean
                                .builder()
                                .skuId(skuId)
                                .originalAmount(new BigDecimal(splitOriginalAmount))
                                .couponReduceAmount(new BigDecimal(splitCouponAmount))
                                .activityReduceAmount(new BigDecimal(splitActivityAmount))
                                .orderAmount(new BigDecimal(splitTotalAmount))
                                .ts(ts)
                                .build();
                    }
                }
        );


        //TODO 5.按照维度sku分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = beanSD.keyBy(TradeSkuOrderBean::getSkuId);


        //TODO 6.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS =
                skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));


        //TODO 7.聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean acc, TradeSkuOrderBean value2) throws Exception {

                        acc.setOriginalAmount(acc.getOriginalAmount().add(value2.getOriginalAmount()));
                        acc.setCouponReduceAmount(acc.getCouponReduceAmount().add(value2.getCouponReduceAmount()));
                        acc.setActivityReduceAmount(acc.getActivityReduceAmount().add(value2.getActivityReduceAmount()));
                        acc.setOrderAmount(acc.getOrderAmount().add(value2.getOrderAmount()));

                        return acc;
                    }
                },
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean orderBean = input.iterator().next();

                        String stt = DateFormatUtil.tsToDateTime(window.getStart());
                        String edt = DateFormatUtil.tsToDateTime(window.getEnd());
                        String curDate = DateFormatUtil.tsToDate(window.getStart());

                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setCurDate(curDate);

                        out.collect(orderBean);
                    }
                }
        );


        //TODO 8.关联sku维度
        //维度关联最基本的实现（每来一条数据都要和hbase关联，效率较低）
        /*
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

                    private Connection hBaseConnection;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConnection = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConn(hBaseConnection);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //根据流中对象，获取要关联的维度主键
                        String skuId = orderBean.getSkuId();

                        //根据维度主键获取对应的唯独对象
                        JSONObject dimSkuInfo =
                                HBaseUtil.getRow(hBaseConnection, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);

                        //补全当前流中对象
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        orderBean.setSkuName(dimSkuInfo.getString("sku_name"));
                        orderBean.setSpuId(dimSkuInfo.getString("spu_id"));
                        orderBean.setTrademarkId(dimSkuInfo.getString("tm_id"));
                        orderBean.setCategory3Id(dimSkuInfo.getString("category3_id"));

                        return orderBean;
                    }
                }
        );
        withSkuInfoDS.print();
        */
        //优化一 : 旁路缓存
        /*
        思路: 先从缓存中获取维度数据，如果在缓存中获取到维度数据，直接将其作为返回值进行返回(缓存命中)
             如果在缓存中没有找到要关联的维度，发送请求到Hbase中获取维度数据，并将从hbase中查询到的维度数据放到缓存中缓存起来
             方便下次查询使用。
        选型：
            状态      性能更好、维护性差（比如维度发生变化，需要在DimApp中删除此处状态，就无法跨程序操作状态了）
            Redis    性能也不错、维护性好     √
        关于Redis中的一些设置
            key:            维度表名:主键值        例如:     dim_base_trademark:1
            value类型:       String
            expire:         1day    避免冷数据常驻内存
        注意：如果业务数据库维度表发生了变化，需要将缓存的维度数据清除掉
        */
        //无模板
        /*
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

                    private Connection hBaseConnection;

                    private Jedis jedis;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConnection = HBaseUtil.getHBaseConnection();
                        jedis = RedisUtil.getJedis();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConn(hBaseConnection);
                        RedisUtil.closeJedis(jedis);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();

                        //从Redis中获取要关联的维度
                        JSONObject dimJsonObj = RedisUtil.readDim(jedis, "dim_sku_info", skuId);
                        if (dimJsonObj != null) {
                            //如果从Redis中获取到了维度数据（缓存命中）
                            System.out.println("从Redis中获取维度数据");
                        } else {
                            //如果没有获取到，发送请求到HBase中查询数据
                            dimJsonObj = HBaseUtil.getRow(hBaseConnection, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                            if (dimJsonObj != null) {
                                System.out.println("从HBase中获取维度数据");
                                //将查询出来的维度放到Redis中缓存起来
                                RedisUtil.writeDim(jedis, "dim_sku_info", skuId, dimJsonObj);
                            } else {
                                System.out.println("没有找到要关联的维度数据");
                            }
                        }

                        //将维度对象相关的属性补充到流中对象上
                        if (dimJsonObj != null) {
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        }
                        return orderBean;
                    }
                }
        );
        //withSkuInfoDS.print();
        */
        //使用模板
        /*
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(new DimMapFunction<TradeSkuOrderBean>() {
            @Override
            public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
            }

            @Override
            public String getTableName() {
                return "dim_sku_info";
            }

            @Override
            public String getRowKey(TradeSkuOrderBean obj) {
                return obj.getSkuId();
            }
        });
        //withSkuInfoDS.print();
        */
        //优化二 : 旁路缓存 + 异步IO（针对单个并行度得异步）
        /*
        1.将异步 I/O 操作应用于 DataStream 作为 DataStream 的一次转换操作
        2.实现分发请求的 AsyncFunction
        3.获取数据库交互的结果并发送给 ResultFuture 的回调函数
        */
        //无模板
        /*
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

                    private AsyncConnection asyncHbaseConn;

                    private StatefulRedisConnection<String, String> asyncRedisConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        asyncHbaseConn = HBaseUtil.getAsyncHBaseConnection();
                        asyncRedisConn = RedisUtil.getAsyncRedisConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeAsyncHBaseConn(asyncHbaseConn);
                        RedisUtil.closeAsyncRedisConnection(asyncRedisConn);
                    }

                    @Override
                    public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();

                        //从Redis中获取维度数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(asyncRedisConn, "dim_sku_info", skuId);

                        if (dimJsonObj != null) {
                            //如果获取到了，缓存命中
                            System.out.println("从Redis中获取数据");
                        } else {
                            //如果没有找到，发送请求到HBase中查询数据
                            dimJsonObj = HBaseUtil.getRowAsync(asyncHbaseConn, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId);
                            if (dimJsonObj != null) {
                                //将查询出来的维度数据缓存到Redis，方便下次查询
                                System.out.println("从HBase中获取数据");
                                RedisUtil.writeDimAsync(asyncRedisConn, "dim_sku_info", skuId, dimJsonObj);
                            } else {
                                System.out.println("从HBase中没有找到数据");

                            }

                        }

                        //将维度属性补充到流中的对象上
                        if (dimJsonObj != null) {
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        }

                        //需要集合类型，转换
                        resultFuture.complete(Collections.singleton(orderBean));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        withSkuInfoDS.print();
        */
        //使用模板
        /**
         * unorderedWait/orderedWait    无序/有序
         *
         * @param DataStream<IN> in             流
         * @param AsyncFunction<IN, OUT> func   如何发送异步请求
         * @param long timeout                  超时时间
         * @param TimeUnit timeUnit             时间单位
         */
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean obj) {
                        return obj.getSkuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //withSkuInfoDS.print();


        //TODO 9.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
                withSkuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSpuName(dimJsonObj.getString("spu_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_spu_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean obj) {
                        return obj.getSpuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );


        //TODO 10.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTrademarkDS = AsyncDataStream.unorderedWait(
                withSpuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setTrademarkName(dimJsonObj.getString("tm_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_trademark";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean obj) {
                        return obj.getTrademarkId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );


        //TODO 11.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withC3DS = AsyncDataStream.unorderedWait(
                withTrademarkDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category3";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory3Name(dim.getString("name"));
                        bean.setCategory2Id(dim.getString("category2_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );


        //TODO 12.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withC2DS = AsyncDataStream.unorderedWait(
                withC3DS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category2";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory2Name(dim.getString("name"));
                        bean.setCategory1Id(dim.getString("category1_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );


        //TODO 13.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withC1DS = AsyncDataStream.unorderedWait(
                withC2DS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category1";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory1Name(dim.getString("name"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );
        //withC1DS.print();

        
        //TODO 14.关联结果写入Doris
        withC1DS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_sku_order_window"));

    }

}

